package com.gitee.xhs.window.evictors;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue;

import java.util.Iterator;

/**
 * @author BigHead
 * @date 2022/7/27 10:36
 * @description AbstractCustomEvictors
 */
public abstract class AbstractCustomEvictors implements Evictor<Transaction, TimeWindow> {

    private final int maxDataSize;

    AbstractCustomEvictors(int maxDataSize){
        this.maxDataSize = maxDataSize;
    }

    protected void removeElement(Iterable<TimestampedValue<Transaction>> elements, int size) {
        if (size > maxDataSize) {
            int removeDataNum = 0;
            for (Iterator<TimestampedValue<Transaction>> iterator = elements.iterator();
                 iterator.hasNext(); ) {
                iterator.next();
                removeDataNum++;
                if (size - removeDataNum < maxDataSize) {
                    break;
                } else {
                    iterator.remove();
                }
            }
        }
    }

}
